package org.example;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Hello world!
 */
public class App {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setParallelism(1);
        StreamTableEnvironment tableEnvironment1 = StreamTableEnvironment.create(environment);


        String createTable = "  CREATE TABLE source_table(\n" +
                "                company_id STRING ,\n" +
                "                uniform_code STRING ,\n" +
                "                loc_city STRING ,\n" +
                "                company_name STRING \n" +
                "        )\n" +
                "        WITH(\n" +
                "                'connector' ='http',\n" +
                "                'format' ='json',\n" +
                "                'url' ='http://10.0.6.30:9200/app_company_zone/_search',\n" +
                "                'type' ='POST' ,\n" +
                "                'union_type' ='Mysql' ,\n" +
                "                'union_url' ='jdbc:mysql://10.0.6.42:3306/xxl_job' ,\n" +
                "                'union_data' ='username'\n" +
                "        )";
        tableEnvironment1.executeSql(createTable);

        Table clickTable = tableEnvironment1.from("source_table");

        tableEnvironment1.createTemporaryView("clickTable",clickTable);
        Table table = tableEnvironment1.sqlQuery("select *from clickTable limit 100 ");

        tableEnvironment1.toDataStream(table).print();



        environment.execute();
    }
}
